package org.myorg.datastraemapi;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.myorg.datastraemapi.model.Student;

import java.util.Properties;

public class KafkaDsProducer {
	public static final String broker_list = "172.18.1.87:9092";
	public static final String topic = "student";  // kafka topic，Flink 程序中需要和这个统一

	public static void writeToKafka() throws InterruptedException {
		Properties props = new Properties();
		props.put("bootstrap.servers", broker_list);
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //key 序列化
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //value 序列化
		KafkaProducer producer = new KafkaProducer<String, String>(props);

		for (int i = 1; i < 100; i++) {
			Student student = new Student(i, i);
			if (i < 50) {
				student.age = 50;
			}

			ProducerRecord<String, String> record = new ProducerRecord(
					topic, null, null, JSON.toJSONString(student));
			producer.send(record, new Callback() {
				@Override
				public void onCompletion(RecordMetadata recordMetadata, Exception e) {
					System.out.println("发送数据: " + JSON.toJSONString(student));
				}
			});

		}

		producer.flush();
	}

	public static void main(String[] args) throws InterruptedException {
		writeToKafka();
	}
}
